package com.atguigu.day03;

import com.atguigu.bean.Event;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Flink08_TransForm_Reduce_Exec {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //2.从端口读取数据
//        DataStreamSource<String> streamSource = env.socketTextStream("localhost", 9999);
        DataStreamSource<Event> streamSource = env.addSource(new Flink01_Source_Customer.ClickSource());

        //3.将从端口读出来的数据转为JavaBean
//        SingleOutputStreamOperator<Event> eventStream = streamSource.map(new MapFunction<String, Event>() {
//            @Override
//            public Event map(String value) throws Exception {
//                String[] split = value.split(",");
//                return new Event(split[0], split[1], Long.parseLong(split[2]));
//            }
//        });

        //TODO 们将数据流按照user进行分区，然后用一个reduce算子实现sum的功能，统计每个用户访问的频次；进而将所有统计结果分到一组，用另一个reduce算子实现maxBy的功能，记录所有用户中访问频次最高的那个，也就是当前访问量最大的用户是谁。
        //将数据转为Tuple2元组
        SingleOutputStreamOperator<Tuple2<String, Integer>> userToOneStream = streamSource.map(new MapFunction<Event, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(Event value) throws Exception {
                return Tuple2.of(value.user, 1);
            }
        });

        //将相同user的数据聚合到一块
        KeyedStream<Tuple2<String,Integer>, String> userKeyedStream = userToOneStream.keyBy(new KeySelector<Tuple2<String,Integer>, String>() {
            @Override
            public String getKey(Tuple2<String,Integer> value) throws Exception {
                return value.f0;
            }
        });

        //使用reduce实现sum的功能
        SingleOutputStreamOperator<Tuple2<String, Integer>> sumResultStream = userKeyedStream.reduce(new ReduceFunction<Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
                return Tuple2.of(value1.f0, value1.f1 + value2.f1);
            }
        });

        //将所有的统计结果放到同一个分组,然后使用reduce实现maxBy的功能求当前访问量最大的用户
        sumResultStream.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                return "a";
            }
        })
                .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
                        return Tuple2.of(value1.f1>value2.f1 ? value1.f0:value2.f0, Math.max(value1.f1, value2.f1));
                    }
                })
                .print("当前访问量最大的用户");


        env.execute();
    }
}
